package io.kiki.sba.registry.server.data_store.multi.cluster.client.handler;

import io.kiki.sba.registry.api.Channel;
import io.kiki.sba.registry.common.model.Node.NodeType;
import io.kiki.sba.registry.common.model.sessionserver.DataChangeRequest;
import io.kiki.sba.registry.server.data_store.bootstrap.ServerConfig;
import io.kiki.sba.registry.server.data_store.multi.cluster.executor.MultiClusterExecutorManager;
import io.kiki.sba.registry.server.data_store.multi.cluster.slot.MultiClusterSlotManager;
import io.kiki.sba.registry.server.shared.remoting.AbstractClientHandler;
import io.kiki.sba.registry.util.StringUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.Set;
import java.util.concurrent.Executor;


public class RemoteDataChangeNotifyHandler extends AbstractClientHandler<DataChangeRequest> {

    private static final Logger logger = LoggerFactory.getLogger(RemoteDataChangeNotifyHandler.class);

    @Autowired
    private ServerConfig serverConfig;

    @Autowired
    private MultiClusterSlotManager multiClusterSlotManager;

    @Autowired
    private MultiClusterExecutorManager multiClusterExecutorManager;


    @Override
    public Class interest() {
        return DataChangeRequest.class;
    }

    @Override
    protected NodeType getConnectNodeType() {
        return NodeType.data_store;
    }

    @Override
    public Executor getExecutor() {
        return multiClusterExecutorManager.getRemoteDataChangeRequestExecutor();
    }

    @Override
    public void checkParam(DataChangeRequest request) {
        StringUtil.checkNotBlank(request.getDataCenter(), "request.dataCenter");
        StringUtil.checkNotEmpty(request.getDataInfoIdToDatumVersionMap(), "request.dataInfoIds");
    }


    @Override
    public Object doHandle(Channel channel, DataChangeRequest request) {

        if (serverConfig.isLocalDataCenter(request.getDataCenter())) {
            logger.error("[DataChangeRequest]local dataCenter data change should not notify localData, request:{}", request);
            return null;
        }
        Set<String> dataInfoIds = request.getDataInfoIdToDatumVersionMap().keySet();
        logger.info("[DataChangeRequest]dataCenter:{} data change:{}", request.getDataCenter(), dataInfoIds.size());
        multiClusterSlotManager.dataChangeNotify(request.getDataCenter(), dataInfoIds);
        return null;
    }
}
